package cc.toprank.syncdatatask.canal;

import cc.toprank.syncdatatask.pojo.MsgInfo;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.lang.SystemUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * description 监听启动&注入rabbit
 * User: WeiDa
 * Date: 2017-5-7
 * Time: 16:56
 */
@Component
public class AbstractCanalClient {

    protected final static Logger logger = LoggerFactory.getLogger(AbstractCanalClient.class);
    protected static final String SEP = SystemUtils.LINE_SEPARATOR;
    protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    protected volatile boolean running = false;
    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {

        public void uncaughtException(Thread t, Throwable e) {
            logger.error("parse events has an error", e);
        }
    };
    protected Thread thread = null;
    protected CanalConnector connector;
    protected static String context_format = null;
    protected static String row_format = null;
    protected static String transaction_format = null;
    protected String destination;
    RabbitTemplate rabbitTemplate;
    protected String dbName;

    protected static String DELETE_SYNC = "o_orderdetailinfo,p_attachmentinfo,p_attachmentmapping";

    protected static final String NO_SYNC = "f_paybankinfo,f_bankcarddealinfo,f_frozeninfo,f_withdrawalsettinginfo,f_withdrawalbankcardinfo,f_withdrawaldealinfo,f_withdrawalinfo,f_paydealinfo";
    static {
        context_format = SEP + "****************************************************" + SEP;
        context_format += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
        context_format += "* Start : [{}] " + SEP;
        context_format += "* End : [{}] " + SEP;
        context_format += "****************************************************" + SEP;

        row_format = SEP
                + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms"
                + SEP;

        transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP;

    }

    public AbstractCanalClient() {
    }

    public AbstractCanalClient(String destination) {
        this(destination, null);
    }

    public AbstractCanalClient(String destination, CanalConnector connector) {
        this.destination = destination;
        this.connector = connector;
    }
    public AbstractCanalClient(String destination, CanalConnector connector,RabbitTemplate rabbitTemplate) {
        this.destination = destination;
        this.connector = connector;
        this.rabbitTemplate=rabbitTemplate;
    }

    protected void start() {
        Assert.notNull(connector, "connector is null");
        thread = new Thread(new Runnable() {

            public void run() {
                process();
            }
        });

        thread.setUncaughtExceptionHandler(handler);
        thread.start();
        running = true;
    }

    protected void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }

        MDC.remove("destination");
    }

    protected void process() {
        int batchSize = 5 * 1024;
        while (running) {
            try {
                MDC.put("destination", destination);
                connector.connect();
                connector.subscribe();
                while (running) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                         try {
                         Thread.sleep(1000);
                         } catch (InterruptedException e) {
                         }
                    } else {
                        printSummary(message, batchId, size);
                        printEntry(message.getEntries());
                    }

                    connector.ack(batchId); // 提交确认
                    //connector.rollback(batchId); // 处理失败, 回滚数据
                }
            } catch (Exception e) {
                logger.error("process error!", e);
            } finally {
                connector.disconnect();
                MDC.remove("destination");
            }
        }
    }

    private void printSummary(Message message, long batchId, int size) {
        long memsize = 0;
        for (Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        logger.info(context_format, new Object[]{batchId, size, memsize, format.format(new Date()), startPosition,
                endPosition});
    }

    protected String buildPositionForDump(Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
    }

    protected void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;

            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) {
                    TransactionBegin begin = null;
                    try {
                        begin = TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    logger.info(transaction_format,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});
                    logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
                } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
                    TransactionEnd end = null;
                    try {
                        end = TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    logger.info("----------------\n");
                    logger.info(" END ----> transaction id: {}", end.getTransactionId());
                    logger.info(transaction_format,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});
                }

                continue;
            }

            if (entry.getEntryType() == EntryType.ROWDATA) {
                RowChange rowChage = null;
                try {
                    rowChage = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                EventType eventType = rowChage.getEventType();

                logger.info(row_format,
                        new Object[]{entry.getHeader().getLogfileName(),
                                String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                                entry.getHeader().getTableName(), eventType,
                                String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});

                if (eventType == EventType.QUERY ) {
                    logger.info(" sql ----> " + rowChage.getSql() + SEP);
                    continue;
                }

                if (eventType == EventType.ALTER || rowChage.getIsDdl()) {
                    logger.info(" sql ----> " + rowChage.getSql() + SEP);
                    MsgInfo msgInfo = new MsgInfo();
                    msgInfo.setSql(rowChage.getSql());
                    msgInfo.setEventType(eventType);
                    msgInfo.setFlag(true);
                    // 4.发送SAVEDATA_FLA消息到对方，并置MsgNeedSyncData=1
                    rabbitTemplate.convertAndSend("fanout", "", msgInfo);
                    logger.info("server publish OK:");
                    continue;
                }

                for (RowData rowData : rowChage.getRowDatasList()) {
                    String tableKey = entry.getHeader().getSchemaName() + "@" + entry.getHeader().getTableName();
                    if(tableKey.equals("@")) continue;
                    //List<CanalEntry.Column> data = rowData.getAfterColumnsList();
                    //printColumn(data);
                    // 获取当前db名称
                    String dbname = entry.getHeader().getSchemaName();
                    if(!dbname.equalsIgnoreCase(dbName)) continue;
                    // 获取当前表名称
                    String tableName = entry.getHeader().getTableName();
                    //雲端資金不需要同步
                    if(NO_SYNC.toLowerCase().contains(tableName.toLowerCase())){
                        continue;
                    }
                    if (eventType == EventType.DELETE) {
                        if(tableName.startsWith("u_") || tableName.startsWith("U_") || DELETE_SYNC.contains(tableName.toLowerCase())){
                            MsgInfo msgInfo = new MsgInfo();
                            Column column = rowData.getBeforeColumns(0);
                            if (column == null) continue;
                            if (column.getIsKey()){
                                msgInfo.setPrimaryKey(column.getName());
                                msgInfo.setIdValue(column.getValue());
                            }
                            String key = tableName + "@" + msgInfo.getIdValue();
                            msgInfo.setMsgId(key);

                            // 3.准备好消息，发送给对方更新并同步数据
                            msgInfo.setDbName(dbname);
                            msgInfo.setTableName(tableName);
                            msgInfo.setEventType(eventType);
                            msgInfo.setFlag(true);
                            // 4.发送SAVEDATA_FLA消息到对方，并置MsgNeedSyncData=1
                            rabbitTemplate.convertAndSend("fanout", "", msgInfo);
                            logger.info("server publish OK:");
                        }
                        //printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT || eventType == EventType.UPDATE) {
                        if("f_fundchangeinfo".equalsIgnoreCase(tableName.toLowerCase())){
                            logger.info("");
                        }

                        if(tableName.equalsIgnoreCase("SystemMaxID")) continue;

                        Column column = rowData.getAfterColumns(0);
                        if (column == null) continue;
                        MsgInfo msgInfo = new MsgInfo();
                        if (column.getIsKey()){
                            msgInfo.setPrimaryKey(column.getName());
                            msgInfo.setIdValue(column.getValue());
                        }
                        String key = tableName + "@" + msgInfo.getIdValue();
                        msgInfo.setMsgId(key);

                        // 3.准备好消息，发送给对方更新并同步数据
                        msgInfo.setDbName(dbname);

                        msgInfo.setTableName(tableName);
                        msgInfo.setEventType(eventType);
                        msgInfo.setColumnList(rowData.getAfterColumnsList());
                        msgInfo.setFlag(true);
                        // 4.发送SAVEDATA_FLA消息到对方，并置MsgNeedSyncData=1
                        rabbitTemplate.convertAndSend("fanout", "", msgInfo);
                        logger.info("server publish OK:");
                    }else {
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    protected void printColumn(List<Column> columns) {
        for (Column column : columns) {
            StringBuilder builder = new StringBuilder();
            builder.append(column.getName() + " : " + column.getValue());
            builder.append("    type=" + column.getMysqlType());
            if (column.getUpdated()) {
                builder.append("    update=" + column.getUpdated());
            }
            builder.append(SEP);
            logger.info(builder.toString());
        }
    }

    public void setConnector(CanalConnector connector) {
        this.connector = connector;
    }

    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void setDbName(String dbName) {
        this.dbName = dbName;
    }

}
